package org.apache.rocketmq.proxy.grpc.v2.client;

import apache.rocketmq.v2.*;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import io.netty.channel.Channel;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.client.*;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.common.channel.ChannelHelper;
import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
import org.apache.rocketmq.proxy.grpc.v2.ContextStreamObserver;
import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcClientChannel;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcProxyException;
import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayResult;
import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.CMResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.filter.FilterAPI;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/13
 * @方法描述：处理客户端操作的活动器
 */
public class ClientActivity extends AbstractMessingActivity {

    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);

    //构造方法
    public ClientActivity(MessagingProcessor messagingProcessor,
                          GrpcClientSettingsManager grpcClientSettingsManager,
                          GrpcChannelManager grpcChannelManager) {
        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
        this.init();
    }

    protected void init() {
        //注册消费者客户端Id监听器
        this.messagingProcessor.registerConsumerListener(new ConsumerIdsChangeListenerImpl());
        //注册生产者客户端Id监听器
        this.messagingProcessor.registerProducerListener(new ProducerChangeListenerImpl());
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理客户端发送过来的心跳请求的方法
     */
    public CompletableFuture<HeartbeatResponse> heartbeat(ProxyContext ctx, HeartbeatRequest request) {
        CompletableFuture<HeartbeatResponse> future = new CompletableFuture<>();

        try {
            //根据上下文得到当前发送请求的客户端在proxy缓存的客户端设置信息
            Settings clientSettings = grpcClientSettingsManager.getClientSettings(ctx);
            //如果客户端设置信息为空，那就给客户端返回一个错误响应，表示没有找到对应的客户端设置信息
            if (clientSettings == null) {
                future.complete(HeartbeatResponse.newBuilder()
                        .setStatus(ResponseBuilder.getInstance().buildStatus(Code.UNRECOGNIZED_CLIENT_TYPE, "cannot find client settings for this client"))
                        .build());
                return future;
            }
            //如果找到了，那就按照客户端类型进行不同的处理
            switch (clientSettings.getClientType()) {
                case PRODUCER: {
                    //执行到这里意味着当前发送请求过来的是生产者客户端
                    //在这里得到并遍历生产者可以发送消息的所有主题
                    for (Resource topic : clientSettings.getPublishing().getTopicsList()) {
                        String topicName = topic.getName();
                        //把每一个主题都注册到生产者管理器中
                        //注意，这里我要多解释一句，客户端发送心跳请求到服务端，这会让服务端每次都执行这个registerProducer方法
                        //在该方法执行的过程中，会刷新为主题创建的ClientChannelInfo客户端信息对象的lastUpdateTimestamp成员变量，这个操作非常重要
                        //会在ProducerManager的scanNotActiveChannel方法中发挥重要作用
                        this.registerProducer(ctx, topicName);
                    }
                    break;
                }
                case PUSH_CONSUMER:
                case SIMPLE_CONSUMER: {
                    //执行到这里就意味着当前发送心跳请求过来的是消费者客户端
                    //在这里验证消费者组的有效性
                    validateConsumerGroup(request.getGroup());
                    String consumerGroup = request.getGroup().getName();
                    //把消费者组注册到消费者管理器中
                    this.registerConsumer(ctx, consumerGroup, clientSettings.getClientType(), clientSettings.getSubscription().getSubscriptionsList(), false);
                    break;
                }
                default: {
                    //如果客户端类型没有被识别，那就回复客户端一个错误相应
                    future.complete(HeartbeatResponse.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.UNRECOGNIZED_CLIENT_TYPE, clientSettings.getClientType().name()))
                            .build());
                    return future;
                }
            }
            //执行到这里意味着处理心跳请求成功了，返回客户端一个成功响应即可
            future.complete(HeartbeatResponse.newBuilder()
                    .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
                    .build());
            return future;
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return future;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理客户端发送的通知客户端终止工作请求的方法
     */
    public CompletableFuture<NotifyClientTerminationResponse> notifyClientTermination(ProxyContext ctx,
                                                                                      NotifyClientTerminationRequest request) {
        CompletableFuture<NotifyClientTerminationResponse> future = new CompletableFuture<>();

        try {
            //得到客户端对应的Id
            String clientId = ctx.getClientID();
            //得到编程语言类型
            LanguageCode languageCode = LanguageCode.valueOf(ctx.getLanguage());
            //从客户端设置信息管理器中得到对应的客户端设置信息
            Settings clientSettings = grpcClientSettingsManager.removeAndGetClientSettings(ctx);

            //接下来就是根据不同的客户端类型执行不同的操作了，其实操作也差不多，那就是把客户端信息注销了
            switch (clientSettings.getClientType()) {
                case PRODUCER:
                    //如果发送请求过来的是生产者客户端
                    for (Resource topic : clientSettings.getPublishing().getTopicsList()) {
                        //那就在这里遍历生产者客户端可以发送消息的所有主题
                        String topicName = topic.getName();
                        //先从客户端channel表中移除对应的客户端channel
                        GrpcClientChannel channel = this.grpcChannelManager.removeChannel(clientId);
                        if (channel != null) {
                            //客户端channel存在的话就创建客户端channel信息对象
                            ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal());
                            //然后把客户端信息从生产管理器中注销了
                            this.messagingProcessor.unRegisterProducer(ctx, topicName, clientChannelInfo);
                        }
                    }
                    break;
                case PUSH_CONSUMER:
                case SIMPLE_CONSUMER:
                    //下面就是注销消费者客户端信息的操作，逻辑和上面分支中的逻辑几乎一致，所以我就不添加注释了，大家自己看看就行
                    validateConsumerGroup(request.getGroup());
                    String consumerGroup = request.getGroup().getName();
                    GrpcClientChannel channel = this.grpcChannelManager.removeChannel(clientId);
                    if (channel != null) {
                        ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal());
                        this.messagingProcessor.unRegisterConsumer(ctx, consumerGroup, clientChannelInfo);
                    }
                    break;
                default:
                    //客户端类型不明确就返回一个错误响应
                    future.complete(NotifyClientTerminationResponse.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.UNRECOGNIZED_CLIENT_TYPE, clientSettings.getClientType().name()))
                            .build());
                    return future;
            }
            //执行到这里意味着请求处理器成功，返回成功响应即可
            future.complete(NotifyClientTerminationResponse.newBuilder()
                    .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
                    .build());
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return future;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：proxy服务端处理客户端发送过来的TelemetryCommand遥测请求的方法
     */
    public ContextStreamObserver<TelemetryCommand> telemetry(StreamObserver<TelemetryCommand> responseObserver) {
        //返回一个流式观察者对象
        return new ContextStreamObserver<TelemetryCommand>() {
            //这个流式观察者的onNext方法会在GrpcMessagingApplication对象的telemetry方法中被调用
            @Override
            public void onNext(ProxyContext ctx, TelemetryCommand request) {
                try {
                    //根据TelemetryCommand遥测请求中的命令执行不同的操作
                    switch (request.getCommandCase()) {
                        case SETTINGS:
                            //如果是设置命令，那就在这里把客户端的设置信息返回给客户端
                            processAndWriteClientSettings(ctx, request, responseObserver);
                            break;
                        case THREAD_STACK_TRACE:
                            //如果是线程堆栈跟踪命令，这意味着什么朋友们？这就意味着是broker节点向proxy发送了想要获得客户端运行信息的请求
                            //proxy向客户端转发了请求，客户端把自己的运行信息收集起来后返回给类proxy，也就是说，这个TelemetryCommand遥测请求中
                            //封装的就是客户端的运行信息，那接下来就要处理这些运行信息了
                            reportThreadStackTrace(ctx, request.getStatus(), request.getThreadStackTrace());
                            break;
                        case VERIFY_MESSAGE_RESULT:
                            //这个就是broker发送的CONSUME_MESSAGE_DIRECTLY请求的结果，客户端把结果回复过来了
                            //原理和上面的分支一致
                            reportVerifyMessageResult(ctx, request.getStatus(), request.getVerifyMessageResult());
                            break;
                    }
                } catch (Throwable t) {
                    //发生异常则处理异常
                    processTelemetryException(request, t, responseObserver);
                }
            }


            @Override
            public void onError(Throwable t) {
                log.error("telemetry on error", t);
            }


            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理遥测请求异常情况的方法
     */
    protected void processTelemetryException(TelemetryCommand request, Throwable t,
                                             StreamObserver<TelemetryCommand> responseObserver) {
        //创建错误状态对象
        StatusRuntimeException exception = io.grpc.Status.INTERNAL
                //设置异常描述信息
                .withDescription("process client telemetryCommand failed. " + t.getMessage())
                //设置异常出现原因
                .withCause(t)
                //转换为运行时异常对象
                .asRuntimeException();
        if (t instanceof GrpcProxyException) {
            GrpcProxyException proxyException = (GrpcProxyException) t;
            //这里就是判断一下程序运行中出现的异常是否在一定范围内
            //而这个一定范围范围指的就是和客户端相关的错误，也就是在Code.BAD_REQUEST_VALUE和Code.INTERNAL_ERROR_VALUE之间
            if (proxyException.getCode().getNumber() < Code.INTERNAL_ERROR_VALUE &&
                    proxyException.getCode().getNumber() >= Code.BAD_REQUEST_VALUE) {
                //如果是和客户端相关的错误，那就把错误状态对象设置为INVALID_ARGUMENT，表示无效参数
                exception = io.grpc.Status.INVALID_ARGUMENT
                        .withDescription("process client telemetryCommand failed. " + t.getMessage())
                        .withCause(t)
                        .asRuntimeException();
            }
        }
        //判断异常类型是不是服务器内部错误
        if (exception.getStatus().getCode().equals(io.grpc.Status.Code.INTERNAL)) {
            //如果是则记录告警日志
            log.warn("process client telemetryCommand failed. request:{}", request, t);
        }
        //把异常结果返回客户端
        responseObserver.onError(exception);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理客户端发送过来的设置命令的TelemetryCommand请求
     */
    protected void processAndWriteClientSettings(ProxyContext ctx, TelemetryCommand request,
                                                 StreamObserver<TelemetryCommand> responseObserver) {

        GrpcClientChannel grpcClientChannel = null;
        //从客户端发送的遥测请求中得到客户端的设置信息，注意，如果是客户端和服务端刚建立连接，客户端第一次向服务端发送遥测请求时
        //那么客户端的设置信息是空的，因为客户端的设置信息实际上是从服务端的配置文件中得到的，这一点在grpcClientSettingsManager对象的mergeProducerData方法中已经得到印证了
        Settings settings = request.getSettings();
        switch (settings.getPubSubCase()) {
            //进入到这个分支就意味着是生产者客户端发送请求过来了
            case PUBLISHING:
                //遍历生产者想要发送消息的所有主题信息
                for (Resource topic : settings.getPublishing().getTopicsList()) {
                    //要整主题信息合法性
                    validateTopic(topic);
                    //得到主题名称
                    String topicName = topic.getName();
                    //把当前主题注册到生产者管理器中，在注册的过程中，会为当前发送请求过来的客户端
                    //创建grpcchannel，当然，如果已经创建了就不必重复创建
                    grpcClientChannel = registerProducer(ctx, topicName);
                    //在这里把流式观察者设置到grpcchannel中，这样一来服务端就可以使用这个grpcchannel主动和客户端进行通信了
                    grpcClientChannel.setClientObserver(responseObserver);
                }
                break;
            case SUBSCRIPTION:
                //执行到这里就意味着发送请求过来的是消费者客户端
                //验证消费者组的合法性
                validateConsumerGroup(settings.getSubscription().getGroup());
                //得到消费组名称
                String groupName = settings.getSubscription().getGroup().getName();
                //把消费者客户端信息注册到消费者管理器中
                grpcClientChannel = registerConsumer(ctx, groupName, settings.getClientType(), settings.getSubscription().getSubscriptionsList(), true);
                //在这里把流式观察者设置到grpcchannel中
                grpcClientChannel.setClientObserver(responseObserver);
                break;
            default:
                break;
        }
        //判断客户端设置信息中是否设置了发布或者订阅信息
        if (Settings.PubSubCase.PUBSUB_NOT_SET.equals(settings.getPubSubCase())) {
            //如果没有设置则返回客户端错误响应
            responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT
                    .withDescription("there is no publishing or subscription data in settings")
                    .asRuntimeException());
            return;
        }
        //这里就是处理客户端设置信息的操作，在这一步会从服务端的配置信息中得到对应的客户端设置信息
        //然后设置到TelemetryCommand对象中，这个TelemetryCommand对象就会返回给客户端，让客户端更新自己本地的设置信息
        TelemetryCommand command = processClientSettings(ctx, request);
        //判断grpcchannel是否不为空
        if (grpcClientChannel != null) {
            //不为空则直接使用grpcchannel的writeTelemetryCommand方法把消息回复给客户端
            grpcClientChannel.writeTelemetryCommand(command);
        } else {
            //这里就是直接返回客户端信息的操作
            responseObserver.onNext(command);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理客户端发送过来的设置信息的方法
     */
    protected TelemetryCommand processClientSettings(ProxyContext ctx, TelemetryCommand request) {
        //得到客户端唯一Id
        String clientId = ctx.getClientID();
        //把客户端设置信息更新到客户端设置信息管理器中
        grpcClientSettingsManager.updateClientSettings(ctx, clientId, request.getSettings());
        //在这里从服务端配置信息中得到客户端的设置信息对象
        Settings settings = grpcClientSettingsManager.getClientSettings(ctx);
        //构建一个TelemetryCommand响应对象，把客户端设置信息返回给客户端
        return TelemetryCommand.newBuilder()
                //设置响应成功状态码
                .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
                //设置客户端要更新的信息
                .setSettings(settings)
                //构建响应对象
                .build();
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把主题信息和对应的客户端信息注册到生产者管理器中的方法
     */
    protected GrpcClientChannel registerProducer(ProxyContext ctx, String topicName) {
        //得到客户端唯一Id
        String clientId = ctx.getClientID();
        //得到语言类型
        LanguageCode languageCode = LanguageCode.valueOf(ctx.getLanguage());
        //在这里为客户端创建对应的grpcchannel对象，如果这个对象已经创建过了，那就不会重复创建，而是直接返回已存在的grpcchannel对象
        GrpcClientChannel channel = this.grpcChannelManager.createChannel(ctx, clientId);
        //创建一个客户端channel信息对象
        ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, parseClientVersion(ctx.getClientVersion()));
        //把主题信息和对应的客户端channel信息注册到生产者管理器中
        this.messagingProcessor.registerProducer(ctx, topicName, clientChannelInfo);
        //以下和事务相关的代码先注释掉
        // TopicMessageType topicMessageType = this.messagingProcessor.getMetadataService().getTopicMessageType(ctx, topicName);
        // if (TopicMessageType.TRANSACTION.equals(topicMessageType)) {
        //     this.messagingProcessor.addTransactionSubscription(ctx, topicName, topicName);
        // }
        return channel;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：注册消费者信息到消费者管理器的方法，该方法在第十版本代码中还用不到，所以我就先不添加注释了，大家也不用关心，等我们把生产者发送消息的功能实现了
     * 并且把消息存储的功能也实现了，然后我就会回过头为大家真正实现消费者客户端以及消费者客户端消费消息的功能，到时候这些方法的注释都会完善
     */
    protected GrpcClientChannel registerConsumer(ProxyContext ctx, String consumerGroup, ClientType clientType,
                                                 List<SubscriptionEntry> subscriptionEntryList, boolean updateSubscription) {
        String clientId = ctx.getClientID();
        LanguageCode languageCode = LanguageCode.valueOf(ctx.getLanguage());
        GrpcClientChannel channel = this.grpcChannelManager.createChannel(ctx, clientId);
        ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, parseClientVersion(ctx.getClientVersion()));
        this.messagingProcessor.registerConsumer(
                ctx,
                consumerGroup,
                clientChannelInfo,
                this.buildConsumeType(clientType),
                MessageModel.CLUSTERING,
                ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET,
                this.buildSubscriptionDataSet(subscriptionEntryList),
                updateSubscription
        );
        return channel;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：解析客户端版本信息的方法
     */
    private int parseClientVersion(String clientVersionStr) {
        int clientVersion = MQVersion.CURRENT_VERSION;
        if (!StringUtils.isEmpty(clientVersionStr)) {
            try {
                String tmp = StringUtils.upperCase(clientVersionStr);
                clientVersion = MQVersion.Version.valueOf(tmp).ordinal();
            } catch (Exception ignored) {
            }
        }
        return clientVersion;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把客户端返回的运行信息的内容上报给Broker节点的方法
     */
    protected void reportThreadStackTrace(ProxyContext ctx, Status status, ThreadStackTrace request) {
        //这个就是responseFuture响应对象的唯一标识
        String nonce = request.getNonce();
        //得到客户端返回的线程堆栈信息
        String threadStack = request.getThreadStackTrace();
        //根据responseFuture的唯一标识从channel管理器中得到对应的responseFuture对象
        CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> responseFuture = this.grpcChannelManager.getAndRemoveResponseFuture(nonce);
        if (responseFuture != null) {
            try {
                //判断客户端响应是否为成功响应
                if (status.getCode().equals(Code.OK)) {
                    //创建消费者运行时信息对象
                    ConsumerRunningInfo runningInfo = new ConsumerRunningInfo();
                    //把消费者客户端的运行时信息都设置到消费者运行时信息对象中
                    runningInfo.setJstack(threadStack);
                    //给responseFuture对象设置成功完成状态，这样一来注册到该对象上的回调方法就会执行，把客户端运行时信息返回给Broker节点了
                    responseFuture.complete(new ProxyRelayResult<>(ResponseCode.SUCCESS, "", runningInfo));
                } else if (status.getCode().equals(Code.VERIFY_FIFO_MESSAGE_UNSUPPORTED)) {
                    responseFuture.complete(new ProxyRelayResult<>(ResponseCode.NO_PERMISSION, "forbidden to verify message", null));
                } else {
                    responseFuture.complete(new ProxyRelayResult<>(ResponseCode.SYSTEM_ERROR, "verify message failed", null));
                }
            } catch (Throwable t) {
                responseFuture.completeExceptionally(t);
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：该方法就不添加注释了，原理和上面的方法一致
     */
    protected void reportVerifyMessageResult(ProxyContext ctx, Status status, VerifyMessageResult request) {
        String nonce = request.getNonce();
        CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> responseFuture = this.grpcChannelManager.getAndRemoveResponseFuture(nonce);
        if (responseFuture != null) {
            try {
                ConsumeMessageDirectlyResult result = this.buildConsumeMessageDirectlyResult(status, request);
                responseFuture.complete(new ProxyRelayResult<>(ResponseCode.SUCCESS, "", result));
            } catch (Throwable t) {
                responseFuture.completeExceptionally(t);
            }
        }
    }


    //以下几个方法在当前版本代码中都用不到，所以我就不添加注释了，大家也没必要现在就关注
    protected ConsumeMessageDirectlyResult buildConsumeMessageDirectlyResult(Status status, VerifyMessageResult request) {

        ConsumeMessageDirectlyResult consumeMessageDirectlyResult = new ConsumeMessageDirectlyResult();
        switch (status.getCode().getNumber()) {
            case Code.OK_VALUE: {
                consumeMessageDirectlyResult.setConsumeResult(CMResult.CR_SUCCESS);
                break;
            }
            case Code.FAILED_TO_CONSUME_MESSAGE_VALUE: {
                consumeMessageDirectlyResult.setConsumeResult(CMResult.CR_LATER);
                break;
            }
            case Code.MESSAGE_CORRUPTED_VALUE: {
                consumeMessageDirectlyResult.setConsumeResult(CMResult.CR_RETURN_NULL);
                break;
            }
        }
        consumeMessageDirectlyResult.setRemark("from gRPC client");
        return consumeMessageDirectlyResult;
    }


    //构建消费者客户端消费模式的方法
    protected ConsumeType buildConsumeType(ClientType clientType) {
        switch (clientType) {
            case SIMPLE_CONSUMER:
                //简单模式客户端主动消费消息
                return ConsumeType.CONSUME_ACTIVELY;
            case PUSH_CONSUMER:
                //push模式客户端被动消费消息
                return ConsumeType.CONSUME_PASSIVELY;
            default:
                throw new IllegalArgumentException("Client type is not consumer, type: " + clientType);
        }
    }


    protected Set<SubscriptionData> buildSubscriptionDataSet(List<SubscriptionEntry> subscriptionEntryList) {
        Set<SubscriptionData> subscriptionDataSet = new HashSet<>();
        for (SubscriptionEntry sub : subscriptionEntryList) {
            String topicName = sub.getTopic().getName();
            FilterExpression filterExpression = sub.getExpression();
            subscriptionDataSet.add(buildSubscriptionData(topicName, filterExpression));
        }
        return subscriptionDataSet;
    }


    protected SubscriptionData buildSubscriptionData(String topicName, FilterExpression filterExpression) {
        String expression = filterExpression.getExpression();
        String expressionType = GrpcConverter.getInstance().buildExpressionType(filterExpression.getType());
        try {
            return FilterAPI.build(topicName, expression, expressionType);
        } catch (Exception e) {
            throw new GrpcProxyException(Code.ILLEGAL_FILTER_EXPRESSION, "expression format is not correct", e);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消费者客户端Id变化监听器，这个监听器在第十版本代码中用不到，所以我就不添加注释了
     */
    protected class ConsumerIdsChangeListenerImpl implements ConsumerIdsChangeListener {


        @Override
        public void handle(ConsumerGroupEvent event, String group, Object... args) {
            switch (event) {
                case CLIENT_UNREGISTER:
                    processClientUnregister(group, args);
                    break;
                case REGISTER:
                    processClientRegister(group, args);
                    break;
                default:
                    break;
            }
        }


        protected void processClientUnregister(String group, Object... args) {
            if (args == null || args.length < 1) {
                return;
            }
            if (args[0] instanceof ClientChannelInfo) {
                ClientChannelInfo clientChannelInfo = (ClientChannelInfo) args[0];
                if (ChannelHelper.isRemote(clientChannelInfo.getChannel())) {
                    return;
                }
                GrpcClientChannel removedChannel = grpcChannelManager.removeChannel(clientChannelInfo.getClientId());
                log.info("remove grpc channel when client unregister. group:{}, clientChannelInfo:{}, removed:{}",
                        group, clientChannelInfo, removedChannel != null);
            }
        }


        protected void processClientRegister(String group, Object... args) {
            if (args == null || args.length < 2) {
                return;
            }
            if (args[1] instanceof ClientChannelInfo) {
                ClientChannelInfo clientChannelInfo = (ClientChannelInfo) args[1];
                Channel channel = clientChannelInfo.getChannel();
                if (ChannelHelper.isRemote(channel)) {
                    Settings settings = GrpcClientChannel.parseChannelExtendAttribute(channel);
                    log.debug("save client settings sync from other proxy. group:{}, channelInfo:{}, settings:{}", group, clientChannelInfo, settings);
                    if (settings == null) {
                        return;
                    }
                    grpcClientSettingsManager.updateClientSettings(
                            ProxyContext.createForInner(this.getClass()),
                            clientChannelInfo.getClientId(),
                            settings
                    );
                }
            }
        }


        @Override
        public void shutdown() {

        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：生产者客户端Id变化监听器
     */
    protected class ProducerChangeListenerImpl implements ProducerChangeListener {


        //处理生产者客户端Id变化的方法
        @Override
        public void handle(ProducerGroupEvent event, String group, ClientChannelInfo clientChannelInfo) {
            //当监听到的事件为客户端注销事件时
            if (event == ProducerGroupEvent.CLIENT_UNREGISTER) {
                //就根据客户端Id从channel管理器中移除对应的channel
                grpcChannelManager.removeChannel(clientChannelInfo.getClientId());
                //从设置管理器中移除对应的客户端设置信息
                grpcClientSettingsManager.removeAndGetRawClientSettings(clientChannelInfo.getClientId());
            }
        }
    }
}
